package org.tao.lightningmq.nameserver.replication;

import com.alibaba.fastjson2.JSON;
import io.netty.channel.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.tao.lightningmq.common.enums.NameServerEventCode;
import org.tao.lightningmq.common.tcp.TcpMsg;
import org.tao.lightningmq.nameserver.common.CommonCache;
import org.tao.lightningmq.nameserver.event.model.SlaveHeartBeatEvent;
import org.tao.lightningmq.nameserver.event.model.StartReplicationEvent;

import java.util.concurrent.TimeUnit;

/**
 * @Author lht
 * @date 2025/1/17 - 15:08
 * @description: 从节点的心跳任务
 */
public class SlaveReplicationHeartBeatTask extends BaseReplicationTask {

    private static final Logger LOGGER = LoggerFactory.getLogger(SlaveReplicationHeartBeatTask.class);

    public SlaveReplicationHeartBeatTask(String taskName) {
        super(taskName);
    }

    @Override
    protected void startTask() {
        try {
            TimeUnit.SECONDS.sleep(2);
            // 先注册从节点到主节点
            StartReplicationEvent startReplicationEvent = new StartReplicationEvent();
            startReplicationEvent.setUser(CommonCache.getNameserverProperties().getNameserverUser());
            startReplicationEvent.setPassword(CommonCache.getNameserverProperties().getNameserverPwd());
            TcpMsg startReplicationMsg = new TcpMsg(NameServerEventCode.START_REPLICATION.getCode(), JSON.toJSONBytes(startReplicationEvent));
            CommonCache.getConnectNodeChannel().writeAndFlush(startReplicationMsg);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }

        // 死循环每3秒发送一次心跳数据
        while (true) {
            try {
                TimeUnit.SECONDS.sleep(3);
                //发送数据给到主节点
                Channel channel = CommonCache.getConnectNodeChannel();
                if (!channel.isActive()) {
                    LOGGER.warn("从节点与主节点连接断开，心跳任务终止");
                    break;
                }
                TcpMsg tcpMsg = new TcpMsg(NameServerEventCode.SLAVE_HEART_BEAT.getCode(), JSON.toJSONBytes(new SlaveHeartBeatEvent()));
                channel.writeAndFlush(tcpMsg);
                if (LOGGER.isDebugEnabled()) {
                    LOGGER.debug("从节点发送心跳数据给master");
                }
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }
    }
}
